Skip to content

compute: columnar Get/Mfp output via as_collection_core - #37763

Merged
antiguru merged 2 commits into
columnar-c2-flatmap-inputfrom
columnar-p1-get-mfp-output
Sep 11, 2026
Merged

compute: columnar Get/Mfp output via as_collection_core#37763
antiguru merged 2 commits into
columnar-c2-flatmap-inputfrom
columnar-p1-get-mfp-output

Conversation

@antiguru

@antiguru antiguru commented Jul 20, 2026

Copy link
Copy Markdown
Member

First producer flip: Get/Mfp emit their output as the columnar edge via as_collection_core, building a ConsolidatingColumnBuilder so within-batch consolidation matches the prior ConsolidatingContainerBuilder.

Columnar dataflow-edge migration. Design doc: doc/developer/design/20260720_columnar_dataflow_edges.md (#37744).

Part of CPU-51.

@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch 2 times, most recently from 8dab0e8 to 2e5920e Compare July 22, 2026 08:41
@linear-code

linear-code Bot commented Jul 22, 2026

Copy link
Copy Markdown

CPU-51

@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 2e5920e to c420261 Compare July 22, 2026 16:24
@antiguru antiguru changed the title compute: columnar Get/Mfp output via as_collection_core (P1) compute: columnar Get/Mfp output via as_collection_core Jul 22, 2026
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from c420261 to 47a405c Compare July 22, 2026 17:50
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch 2 times, most recently from 0a3b022 to 570a36a Compare August 19, 2026 13:17
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 570a36a to b01857d Compare August 20, 2026 08:49
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from b01857d to 176a1c1 Compare August 20, 2026 09:13
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 176a1c1 to 489ab17 Compare September 6, 2026 18:23
@antiguru
antiguru marked this pull request as ready for review September 10, 2026 11:25
@antiguru
antiguru requested a review from a team as a code owner September 10, 2026 11:25
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 489ab17 to 4bcab91 Compare September 10, 2026 11:33
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 4bcab91 to 00417d5 Compare September 10, 2026 12:09
@def-

def- commented Sep 10, 2026

Copy link
Copy Markdown
Contributor

QA LLM Review

1. MEDIUM -- Mfp/Get output messages are no longer capped at ~204 records, coarsening FlatMap's yield granularity

src/compute/src/render/context.rs:1058

ConsolidatingContainerBuilder<Vec<(Row, T, Diff)>> cut every message this producer emitted to at most default_capacity::<(Row, T, Diff)>() = 204 records, while ConsolidatingColumnBuilder mints a container only once the serialized accumulator reaches ~1.8 MiB, so messages are now as large as the producing session can make them. FlatMapStage tests COMPUTE_FLAT_MAP_FUEL only between input messages, so a table function reading a Get/Mfp edge now expands two orders of magnitude more input rows per activation before the worker can yield.

Details

Measured at this commit through as_collection_core, 400,000 single-int64 output rows. Unkeyed path (flat_map_datums, one session per input message): 391 messages of 1024 records, so the producer now passes the input message size through instead of re-cutting it to 204. Keyed path (PendingWork::do_work opens one session and walk_cursor pushes into it until REFUEL = 1,000,000 records, so FLUSH_THRESHOLD_WORDS at src/timely-util/src/columnar/consolidate.rs:62 is what bounds the message): 6 messages of ~67,400 records, against ~1,960 messages of 204 before, a 330x jump. A 1 KiB row still gives ~1,800 records per message, 9x. Message::push_at ships a container as one message and never splits it.

flat_map_stage pops one queue entry per input message, runs data.for_each_record(...) to completion (src/compute/src/render/flat_map.rs:170), and only then checks if budget == 0 (:187); neither process_flat_map_row nor drain_through_mfp breaks out of the per-record loop. The operator doc at :50 already states it "can only yield between input batches, but not from within a batch". The unkeyed FlatMap input consumes the edge directly (flat_map.rs:58-66), and message sizes propagate, since a downstream Mfp re-emits at whatever size its input arrived in. One arrangement-scan Get upstream therefore sets the granularity for the rest of the chain: a generate_series(1, 100) behind such an input goes from ~20k output records per activation to ~6.7M in one non-preemptible activation, stalling peeks and frontier advancement for every dataflow sharing that worker.

Two smaller consequences of the same size change: where the edge forks, Tee clones a ~1.8 MiB column per extra consumer rather than a 204-record Vec, and the ColumnarToVec seams (sink, LetRec binding, temporal bucketing) materialize a whole container's worth of owned Rows per activation.

The ~1.8 MiB ship threshold is ColumnBuilder's, and until this PR it was only ever reached on the arrange path (FormArrangementKey into mz_arrange_core), which is built for bulk; no producer emitted CollectionEdge::Columnar. This PR is what puts containers of that size on a general-purpose collection edge in front of arbitrary Pipeline consumers. The within-batch fuel gap in FlatMapStage is itself already tracked in Linear CPU-163; what this diff changes is the size of the batch. Bounding what this producer ships is the narrow fix; letting FlatMapBatch::for_each_record stop and resume at an offset is the durable one, and the Column arm is index-addressable so the resume state is one usize.

@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from 00417d5 to b74bdeb Compare September 10, 2026 18:37
@antiguru

Copy link
Copy Markdown
Member Author

Verified, and not fixed in this stack. Flagging it rather than patching it, because the fix changes an operator scheduling contract and should be a decision rather than a drive-by.

The mechanism checks out. flat_map_stage runs data.for_each_record(...) to completion over one input message and only then tests if budget == 0, so the non-preemptible unit is one message, and this PR is what makes that unit large: ConsolidatingColumnBuilder mints a container at FLUSH_THRESHOLD_WORDS, roughly 1.8 MiB, where ConsolidatingContainerBuilder<Vec<_>> cut one every default_capacity records. The operator doc already records the within-batch gap; what changes here is the size of the batch.

Neither option the review offers is free. Bounding this producer gives back the bulk containers the migration exists for, on every consumer, to fix a granularity problem in one of them. Making FlatMapBatch resumable is the right shape, and the Column arm is index-addressable so the resume state is one usize, but it means the queue entry carries an offset and the operator can stop mid-batch, which changes when capabilities are dropped. That belongs in its own PR with its own test, next to CPU-163 rather than inside the producer flip.

My preference is the resumable batch as a follow-up on top of the stack, the way #38368 came off the join review. Say the word and I will open it; if you would rather bound the producer first as a stopgap, that is a one-line threshold and I can measure what it costs the arrange path.

Posted by Claude Code

Make `as_collection_core` build its ok output into a
`ConsolidatingColumnBuilder` and return a `CollectionEdge::Columnar`,
flipping the Get and Mfp producers to emit the columnar edge. This is the
first producer flip: the columnar arm goes live for Get/Mfp-fed edges,
which Wave 1 already made every consumer accept natively or decode only at
a sanctioned leaf.

Rework the identity fast-path so an unarranged trivial MFP hands the input
edge straight through (a columnar producer stays columnar with no
`ColumnarToVec` hop) instead of forcing `as_specific_collection` to decode
it to `Vec`. A keyed identity still reads its arrangement as a `Vec` edge.
`as_specific_collection` stays the row-based consumer leaf.

`ConsolidatingColumnBuilder` folds within-batch duplicates, matching the
row-based `ConsolidatingContainerBuilder` this replaced, and emits a
`Column` instead of a `Vec`. It stages owned `(Row, T, Diff)` tuples to
consolidate in place, so the producer gives its records owned.
`mfp_plan.evaluate` already produces a fresh owned `Row` per result, so
this is a move into staging, not a new allocation. The borrowed-push
no-owned-Row pattern is a consumer optimization for reading an existing
columnar batch; it never applied to a producer computing new rows.

The internal fueled `flat_map` chain (`CollectionBundle::flat_map`,
`ArrangementFlavor::flat_map`, `flat_map_core_fallible`, `PendingWork::
do_work`) carried a `PushInto<(D, T, Diff)>` bound that constrained the
element type up front. The bound was never exercised by those bodies (the
caller's `logic` performs the `give`), so relax it to `DCB:
ContainerBuilder` and drop the now-unused `D`, letting the producer choose
its own builder and push shape through the same fuel machinery.

Sink and `ensure_collections` temporal bucketing decode the produced edge
at their existing sanctioned leaves.

Tests: a `Get -> ArrangeBy` end-to-end unit test asserting the producer
emits columnar and the arrange input keeps the columnar passthrough; an
identity-fast-path test asserting the input edge variant is preserved; and
a within-batch consolidation test asserting duplicate output rows fold to a
single record with summed diff.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Give the missing-push-bound rationale a single owner, drop the comparisons
against the row-based builder, and cut the test doc to what the asserts do
not already state.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@antiguru
antiguru force-pushed the columnar-p1-get-mfp-output branch from b74bdeb to 45cd71b Compare September 11, 2026 18:52
@antiguru
antiguru merged commit 98ecec8 into main Sep 11, 2026
84 checks passed
@antiguru
antiguru deleted the columnar-p1-get-mfp-output branch September 11, 2026 20:31
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants